package com.example.test.stream17;

import com.example.constant.KafkaTopic;
import com.example.entity.stream17.*;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerABCDEFGHIJExample {


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        Producer<String, A17> producerA = new KafkaProducer<>(props);
        Producer<String, B17> producerB = new KafkaProducer<>(props);
        Producer<String, C17> producerC = new KafkaProducer<>(props);
        Producer<String, D17> producerD = new KafkaProducer<>(props);
        Producer<String, E17> producerE = new KafkaProducer<>(props);
        Producer<String, F17> producerF = new KafkaProducer<>(props);
        Producer<String, G17> producerG = new KafkaProducer<>(props);
        Producer<String, H17> producerH = new KafkaProducer<>(props);
        Producer<String, I17> producerI = new KafkaProducer<>(props);
        Producer<String, J17> producerJ = new KafkaProducer<>(props);
           int userId = 1;

            while (userId < 1000000){
                A17 a = new A17(userId,"a1_"+userId,"a2_"+userId);
                B17 b = new B17(userId,"b1_"+userId,"b2_"+userId);
                C17 c = new C17(userId,"c1_"+userId,"c2_"+userId);
                D17 d = new D17(userId,"d1_"+userId,"d2_"+userId);
                E17 e = new E17(userId,"e1_"+userId,"e2_"+userId);
                F17 f = new F17(userId,"f1_"+userId,"f2_"+userId);
                G17 g = new G17(userId,"g1_"+userId,"g2_"+userId);
                H17 h = new H17(userId,"h1_"+userId,"h2_"+userId);
                I17 i = new I17(userId,"i1_"+userId,"i2_"+userId);
                J17 j = new J17(userId,"j1_"+userId,"j2_"+userId);
                ProducerRecord<String, A17> recordA = new ProducerRecord<>(KafkaTopic.ATopic17, String.valueOf(userId), a);
                ProducerRecord<String, B17> recordB = new ProducerRecord<>(KafkaTopic.BTopic17, String.valueOf(userId), b);
                ProducerRecord<String, C17> recordC = new ProducerRecord<>(KafkaTopic.CTopic17, String.valueOf(userId), c);
                ProducerRecord<String, D17> recordD = new ProducerRecord<>(KafkaTopic.DTopic17, String.valueOf(userId), d);
                ProducerRecord<String, E17> recordE = new ProducerRecord<>(KafkaTopic.ETopic17, String.valueOf(userId), e);
                ProducerRecord<String, F17> recordF = new ProducerRecord<>(KafkaTopic.FTopic17, String.valueOf(userId), f);
                ProducerRecord<String, G17> recordG = new ProducerRecord<>(KafkaTopic.GTopic17, String.valueOf(userId), g);
                ProducerRecord<String, H17> recordH = new ProducerRecord<>(KafkaTopic.HTopic17, String.valueOf(userId), h);
                ProducerRecord<String, I17> recordI = new ProducerRecord<>(KafkaTopic.ITopic17, String.valueOf(userId), i);
                ProducerRecord<String, J17> recordJ = new ProducerRecord<>(KafkaTopic.JTopic17, String.valueOf(userId), j);
                try {
                    producerA.send(recordA);
                    producerB.send(recordB);
                    producerC.send(recordC);
                    producerD.send(recordD);
                    producerE.send(recordE);
                    producerF.send(recordF);
                    producerG.send(recordG);
                    producerH.send(recordH);
                    producerI.send(recordI);
                    producerJ.send(recordJ);
                    System.out.println("Message sent successfully");
                    userId++;
                } catch (Exception exception) {
                    exception.printStackTrace();
                } finally {


                }
            }

        producerA.flush();
        producerB.flush();
        producerC.flush();
        producerD.flush();
        producerE.flush();
        producerF.flush();
        producerG.flush();
        producerH.flush();
        producerI.flush();
        producerJ.flush();

        producerA.close();
        producerB.close();
        producerC.close();
        producerD.close();
        producerE.close();
        producerF.close();
        producerG.close();
        producerH.close();
        producerI.close();
        producerJ.close();

    }

}